package com.pt25.base.dep.kafka;

import com.pt25.base.exception.BizException;
import com.pt25.base.util.StrUtil;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Set;


@Data
@Component
@ConfigurationProperties(prefix = "kafka-configs")
class KafkaProperties {
    /**
     * 默认producer
     */
    private String primary;
    private List<Consumer> consumers;
    private List<Producer> producers;

    boolean checkConsumerConfigs() {
        return checkConfigs(consumers);
    }

    boolean checkProducerConfigs() {
        return checkConfigs(producers);
    }


    @EqualsAndHashCode(callSuper = true)
    @Data
    static class Consumer extends Properties {
        /**
         * 在偏移量无效的情况下，消费者将从起始位置读取分区的记录
         * <p>
         * latest：最新的位置开始消费
         * earliest：最后消费完的位置接着开始消费
         */
        private String autoOffsetReset = "latest";
        /**
         * 自动提交消费偏移量
         */
        private Boolean enableAutoCommit = true;
        /**
         * 自动提交消费的间隔时间 ms
         */
        private Integer autoCommitInterval = 1000;
        /**
         * 是否批量消费
         */
        private Boolean batchListener = true;
        /**
         * 批量消费批次大小
         */
        private Integer batchSize = 10;
        /**
         * 在侦听器容器中运行的线程数
         */
        private Integer concurrency = 10;
        /**
         * 账号
         */
        private String username;
        /**
         * 密码
         */
        private String password;
    }

    @EqualsAndHashCode(callSuper = true)
    @Data
    static class Producer extends Properties {
        /**
         * acks：ack具有3个取值 0、1和-1(all)
         * acks=0：发送消息时，完全不用等待broker的处理结果，即可进行下一次发送，吞吐量最高
         * acks=-1或all：不仅需要leader broker将消息持久化，同时还需要等待ISR副本（in-sync replica 即与leader节点保持消息同步的副本集合）全部都成功持久化消息后，才返回响应结果，吞吐量最低
         * ack=1：是前面两个的折中，只需要leader broker持久化后，便响应结果
         */
        private String acksConfig = "1";
        /**
         * 发送失败重试次数
         */
        private Integer retries = 1;
        /**
         * 账号
         */
        private String username;
        /**
         * 密码
         */
        private String password;

    }

    @Data
    static class Properties {
        /**
         * 集群名称
         */
        private String name;
        /**
         * kafka服务地址
         */
        private String bootstrapServers;
    }

    private <T extends Properties> boolean checkConfigs(List<T> list) {
        Set<String> nameSet = new HashSet<>();
        list.forEach(config -> {
            nameSet.add(config.getName());
            if (StrUtil.isBlank(config.getBootstrapServers())) {
                throw new BizException("bootstrapServers is empty");
            }
            if (StrUtil.isBlank(config.getName())) {
                throw new BizException("name is empty");
            }
        });
        if (nameSet.size() != list.size()) {
            throw new BizException("集群名称有重复！");
        }
        return true;
    }

}
